package storm;

import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Tuple;

/**
 * @author Hanhan_Jun
 * @date 2021年12月13日 10:26
 */
public class Bolt2_1  extends BaseBasicBolt {
    public static JRedisUtil jredis =new JRedisUtil("10.86.184.170",6379);

    public void execute(Tuple input, BasicOutputCollector collector) {
        try {
            String recipient = input.getStringByField("recipient");
            String receiveplace = input.getStringByField("receiveplace");
            Double total = input.getDoubleByField("total");

            if (receiveplace!=null){
                if (receiveplace.contains("内蒙")){
                    receiveplace = receiveplace+"古";
                    jredis.zincrby("receiveplace",1,receiveplace);
                }else if(receiveplace.contains("黑龙")){
                    receiveplace = receiveplace+"江";
                    jredis.zincrby("receiveplace",1,receiveplace);
                }else {
                    jredis.zincrby("receiveplace",1,receiveplace);
                }
            }
            jredis.zincrby("NumOfRecipent",1,recipient+receiveplace);
            jredis.incrBy("Total",total.longValue());
            jredis.incrBy("TotalOrders", 1L);

            System.out.println("Done22222222222222222222222222222222222");
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("ERROR In orderdetail");
        }

    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}